package com.xhg.mqtt.job;

import com.xhg.mqtt.constant.MqttConstants;
import com.xhg.mqtt.constant.MsgConstants;
import com.xhg.mqtt.listener.MessageEvent;
import com.xhg.mqtt.listener.MessageManager;
import com.xhg.mqtt.listener.MqttLoggerFactory;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * 消息重发队列
 *
 * @author E.T
 * @version 1.0
 * @Description
 * @date 2018-07-23 11:02
 * @updateby
 * @updatedate
 * @since 1.0
 */
public class MessageRepeatQueue {

    private Map<String, MessageEvent> map = new ConcurrentHashMap<>();
    private Map<String, MessageEvent> failMsgMap = new ConcurrentHashMap<>();

    private static class MessageRepeatQueueHolder{
        private final static MessageRepeatQueue instance=new MessageRepeatQueue();
    }
    public static MessageRepeatQueue getInstance(){
        return MessageRepeatQueueHolder.instance;
    }

    private MessageRepeatQueue(){
    }

    /**
     * 添加队列
     */
    public void add(String messageId, MessageEvent event) {
        map.put(messageId, event);
    }

    /**
     * 移除队列内容
     */
    public void remove(String messageId) {
        map.remove(messageId);
    }

    /**
     * 获取失败消息队列map
     * @return
     */
    public Map<String, MessageEvent> getFailMsgMap() {
        return failMsgMap;
    }

    /**
     * 清空失败消息队列
     */
    public void removeAllFailMsg(){
        failMsgMap.clear();
    }

    /**
     * 清空重发消息队列
     */
    public void removeAllRecMsg(){
        map.clear();
    }
    /**
     * 清空失败消息队列
     */
    public void removeFailMsg(String messageId){
        failMsgMap.remove(messageId);
    }



    public void init() {
        ScheduledExecutorService service=  Executors.newScheduledThreadPool(MqttConstants.COREPOOLSIZE,
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r,"消息重发线程");
                }
            });
        service.scheduleAtFixedRate(new MessageTimer(map, failMsgMap), MqttConstants.INIT_START, MqttConstants.PERIOD, TimeUnit.SECONDS);
    }
    /**
     * 监听重发3次都发送失败的消息，将这些消息保存到数据库里面，以便后面查询再次重发(目前只处理单条消息)
     * 监听间隔时间可以自己设定
     */
    public void listenerFailMsg(){
        ScheduledExecutorService service= Executors.newSingleThreadScheduledExecutor(
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r,"失败消息处理线程");
                }
            });
        service.scheduleAtFixedRate(()-> {
                Map<String, MessageEvent> failMsgMap = getFailMsgMap();
                if (!failMsgMap.isEmpty()){
                    MqttLoggerFactory.warn(MsgConstants.SAVE_FAIL_MESSAGE +failMsgMap.keySet().toString()+"");
                    //这里保存失败消息，保存完成后清空失败队列
                        String messageId = MessageManager.failMessage(failMsgMap);
                        if (messageId != null) {
                            removeFailMsg(messageId);
                        }
                }else {
                    //没有失败消息退出定时器
                    service.shutdown();
                }
        },MqttConstants.INIT_START,MqttConstants.PERIOD, TimeUnit.SECONDS);
    }
}
